package com.stormdemo.wclocalhost.bolt;

import java.util.HashMap;
import java.util.Map.Entry;

import com.stormdemo.wclocalhost.MapCatch;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class wordCountBolt extends BaseBasicBolt {
	
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}

	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
			String str = input.getString(0);
			if (!MapCatch.parames.containsKey(str)) {
				MapCatch.parames.put(str, Integer.valueOf(1));
			    } else {
			      Integer c = Integer.valueOf(((Integer)MapCatch.parames.get(str)).intValue() + 1);
			      MapCatch.parames.put(str, c);
			    }
			
			System.out.println("======================");
			for(Entry<String, Integer> en:MapCatch.parames.entrySet()){
				System.out.println(en.getKey()+"---"+en.getValue());
			}
			System.out.println("======================");
	}
}
